package org.baojie.raft.future;

import org.baojie.raft.util.current.RaftUnsafe;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.CancellationException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.locks.LockSupport;

// 可以返回自定义对象的future
// 可单独使用
public class RaftFuture<T> {

    private static final Logger log = LoggerFactory.getLogger(RaftFuture.class);

    /**
     * NEW -> DOING
     * NEW -> DONE
     * NEW -> CANCELLED
     */

    private static final int NEW = 0;
    private static final int DOING = 1;
    private static final int DONE = 2;
    private static final int EXCEPTIONAL = 3;
    private static final int CANCELLED = 4;

    private Object outcome;
    private volatile int state;
    private volatile WaitNode waiters;
    private volatile RListener listener;

    private final long notifyTimeoutMillis;

    private RaftFuture(Class<T> kind, long notifyTimeoutMillis) {
        this.state = NEW;
        this.notifyTimeoutMillis = notifyTimeoutMillis;
    }

    public static final <T> RaftFuture create(Class<T> kind, long notifyTimeoutMillis) {
        if (kind == null) {
            throw new NullPointerException();
        } else if (notifyTimeoutMillis < 0) {
            throw new IllegalArgumentException();
        } else {
            return new RaftFuture(kind, notifyTimeoutMillis);
        }
    }

    public boolean isDone() {
        if (state >= DONE) {
            return true;
        } else {
            return false;
        }
    }

    public boolean cancel() {
        try {
            if (state == NEW) {
                if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, CANCELLED)) {
                    return true;
                } else {
                    if (state == CANCELLED) {
                        return true;
                    } else {
                        return false;
                    }
                }
            } else {
                if (state == CANCELLED) {
                    return true;
                } else {
                    return false;
                }
            }
        } finally {
            finishCompletion();
        }
    }

    public RaftFuture<T> addListener(RListener<? extends RaftFuture<? super T>> rl) {
        if (null == rl) {
            throw new NullPointerException();
        } else {
            synchronized (this) {
                if (null != listener) {
                    throw new IllegalStateException();
                } else {
                    listener = rl;
                }
            }
            if (isDone()) {
                done(this);
            }
            return this;
        }
    }

    public T get() throws InterruptedException {
        int s = state;
        if (s <= DOING) {
            s = awaitDone(false, 0L);
        }
        return report(s);
    }

    public T get(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException {
        if (null == unit) {
            throw new NullPointerException();
        }
        int s = state;
        if (s <= DOING && (s = awaitDone(true, unit.toNanos(timeout))) <= DOING) {
            throw new TimeoutException();
        }
        return report(s);
    }

    public final void set(T v) {
        if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, DOING)) {
            try {
                outcome = v;
                UNSAFE.putOrderedInt(this, stateOffset, DONE);
            } finally {
                finishCompletion();
            }
        }
    }

    public final void setException(Throwable t) {
        if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, DOING)) {
            try {
                outcome = t;
                UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
            } finally {
                finishCompletion();
            }
        }
    }

    @SuppressWarnings("unchecked")
    private T report(int s) {
        Object x = outcome;
        if (s == DONE) {
            return (T) x;
        }
        if (s >= CANCELLED) {
            throw new CancellationException();
        }
        throw new Error();
    }

    protected void done(final RaftFuture future) {
        int tmp = state;
        if (tmp == DONE) {
            final RListener rl = listener;
            if (null != rl) {
                final FutureExecutor executor = FutureExecutor.getInstance();
                Runnable run = new Runnable() {

                    @Override
                    public void run() {
                        if (isDone()) {
                            try {
                                notifyListener();
                            } catch (Throwable t) {
                                log.error("raft future do listener error = " + t.getMessage(), t);
                            }
                        }
                    }

                    private void notifyListener() {
                        if (!occurError()) {
                            rl.operationComplete(future);
                        }
                    }

                    private boolean occurError() {
                        final Object tmp = outcome;
                        if (null != tmp && (tmp instanceof Throwable)) {
                            Throwable cause = (Throwable) tmp;
                            log.error("find raft future cause not null, error = " + cause.getMessage(), cause);
                            return true;
                        } else {
                            return false;
                        }
                    }
                };
                executor.submit(run, notifyTimeoutMillis, TimeUnit.MILLISECONDS);
            }
        }
    }

    private static final class WaitNode {
        volatile Thread thread;
        volatile WaitNode next;

        WaitNode() {
            thread = Thread.currentThread();
        }
    }

    private void finishCompletion() {
        try {
            weakUpWaiters();
        } finally {
            done(this);
        }
    }

    private void weakUpWaiters() {
        // 此处for循环为了处理完一批node时候又进来一批node
        // 所以要使用for循环
        for (WaitNode q; (q = waiters) != null; ) {
            // 使用if的cas确保处理此方法的只有一个线程
            if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
                for (; ; ) {
                    Thread t = q.thread;
                    if (t != null) {
                        q.thread = null;
                        LockSupport.unpark(t);
                    }
                    WaitNode next = q.next;
                    if (next == null)
                        break;
                    q.next = null;
                    q = next;
                }
                break;
            }
        }
    }

    private int awaitDone(boolean timed, long nanos) throws InterruptedException {
        final long deadline = timed ? System.nanoTime() + nanos : 0L;
        WaitNode q = null;
        boolean queued = false;
        for (; ; ) {
            // 某个线程中断不影响其他线程
            if (Thread.interrupted()) {
                removeWaiter(q);
                throw new InterruptedException();
            }
            int s = state;
            // 已经完成或者取消直接返回当前状态
            if (s > DOING) {
                if (q != null) {
                    q.thread = null;
                }
                return s;
            } else if (s == DOING) {
                // 此中间状态只有set方法中设置
                // 所以会很快发生变化
                Thread.yield();
            } else if (q == null) {
                q = new WaitNode();
            } else if (!queued) {
                // 将当前节点的next设置成已经入队的成员链
                // 并且将队列指针设置成q
                // 如果成功会继续for然后会进入下面else if
                // 否则再次for循环如队列
                queued = UNSAFE.compareAndSwapObject(this, waitersOffset, q.next = waiters, q);
            } else if (timed) {
                nanos = deadline - System.nanoTime();
                // 如果小于0说明已经超时了或者是非超时的get
                // 那么将已经添加进去的node移除掉
                // 并且返回当前状态
                if (nanos <= 0L) {
                    removeWaiter(q);
                    return state;
                }
                // 否则等待相应的时间长度
                // LockSupport相应中断且不会抛异常
                LockSupport.parkNanos(this, nanos);
            } else {
                // 如果不是超时等待的方式
                // 那么就一致阻塞知道被中断后者被取消
                LockSupport.park(this);
            }
        }
    }

    // 移除已经入队列的node
    private void removeWaiter(WaitNode node) {
        if (node != null) {
            node.thread = null;
            retry:
            for (; ; ) {
                for (WaitNode pred = null, q = waiters, s; q != null; q = s) {
                    s = q.next;
                    if (q.thread != null) {
                        pred = q;
                    } else if (pred != null) {
                        pred.next = s;
                        if (pred.thread == null) {
                            continue retry;
                        }
                    } else if (!UNSAFE.compareAndSwapObject(this, waitersOffset, q, s)) {
                        continue retry;
                    }
                }
                break;
            }
        }
    }

    private static final sun.misc.Unsafe UNSAFE;
    private static final long stateOffset;
    private static final long waitersOffset;

    static {
        try {
            UNSAFE = RaftUnsafe.getUnsafe();
            Class<?> k = RaftFuture.class;
            stateOffset = UNSAFE.objectFieldOffset(k.getDeclaredField("state"));
            waitersOffset = UNSAFE.objectFieldOffset(k.getDeclaredField("waiters"));
        } catch (Throwable e) {
            throw new Error(e);
        }
    }

}
